package com.hjs.homework

import java.util.Properties
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.json4s._
import org.json4s.jackson.JsonMethods._

import scala.io.BufferedSource

class OrderBean(val ts: Long, val orderId: String, val userId: String, val orderStatusId: Int, val orderStatus: String, val payModeId: Int, val payMode: String, val payment: String, val products: List[Product])

class Product(val productId: String, val productName: String, val price: Double, val productNum: Int, val categoryid: String, val catname1: String, val catname2: String, val catname3: String)

object KafkaProducerForDruid {
  def main(args: Array[String]): Unit = {
    // 定义 kafka 参数
//    val brokers = "centos7kafkanode1,centos7kafkanode1,centos7kafkanode1"
    val topic = "druid_homework"
    val prop = new Properties()
    prop.put("bootstrap.servers", "bogon:9092")
    prop.put("zookeeper.connect", "bogon:2181")
    prop.put("group.id", "homework-group")
    prop.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    prop.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
//    prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
//    prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
//    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
//    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
//    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
    val producer = new KafkaProducer[String, String](prop)


    implicit val formats = DefaultFormats

    val source: BufferedSource = scala.io.Source.fromFile("data/orders.json")
    val iter: Iterator[String] = source.getLines()
    iter.foreach { line =>
      val order: OrderBean = parse(line).extract[OrderBean]
      val products = order.products
      for (product <- products) {
        val orderDetail = "{\"ts\":" + order.ts + ",\"orderId\":\"" + order.orderId + "\",\"userId\":\"" + order.userId + "\",\"orderStatusId\":" + order.orderStatusId + "," +
          "\"orderStatus\":\"" + order.orderStatus + "\",\"payModeId\":" + order.payModeId + ",\"payMode\":\"" + order.payMode + "\",\"payment\":\"" + order.payment + "\"," +
          "\"product\":{\"productId\":\"" + product.productId + "\",\"productName\":\"" + product.productName + "\",\"price\":" + product.price + ",\"productNum\":" + product.productNum + "," +
          "\"categoryid\":\"" + product.categoryid + "\",\"catname1\":\"" + product.catname1 + "\",\"catname2\":\"" + product.catname2 + "\",\"catname3\":\"" + product.catname3 + "\"}}"
        //        println(orderDetail)
        val msg = new ProducerRecord[String, String](topic, orderDetail)
        producer.send(msg)
      }
      Thread.sleep(1000)
    }
    producer.close()
    source.close()
  }
}
